package com.aurora.advanced;

import org.apache.flink.streaming.connectors.elasticsearch.ActionRequestFailureHandler;
import org.apache.flink.streaming.connectors.elasticsearch.RequestIndexer;
import org.elasticsearch.action.ActionRequest;
import org.elasticsearch.common.util.concurrent.EsRejectedExecutionException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 描述：es写入异常处理
 *
 * @author 浅夏的猫
 * @version 1.0.0
 * @date 2024-02-13 00:04:24
 */
public class CustomActionRequestFailureHandler implements ActionRequestFailureHandler {

    private static final Logger logger = LoggerFactory.getLogger(CustomActionRequestFailureHandler.class);

    /**
     * 描述：esSink写数据异常或失败处理
     *
     * @param action 操作
     * @param throwable 异常
     * @param restStatusCode 结果返回码
     * @param requestIndexer request操作
     * @throws Throwable
     */
    @Override
    public void onFailure(ActionRequest action, Throwable throwable, int restStatusCode, RequestIndexer requestIndexer) throws Throwable {
        // 处理不同类型的异常
        if (throwable instanceof EsRejectedExecutionException) {
            // 如果是由于线程池饱和导致的拒绝执行异常，可以采取相应的处理措施
            logger.warn("Elasticsearch action execution was rejected due to thread pool saturation.");
            // 这里你可以选择执行重试或者其他处理逻辑，例如将数据写入到一个备用存储
            // 例如： indexer.add(createAnotherRequest(action));
        } else {
            // 对于其他类型的异常，默认返回放弃策略
            logger.error("Unhandled failure, abandoning request: {}", action.toString());
        }
    }
}
